package com.aididi.config;

import com.aididi.utils.SM4Util;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author ：xuzexiang
 * @description：MQ生产者配置
 * @date ： 2021/10/28 9:26
 */
@Configuration
@Slf4j
public class ProducerConfig {

    @Bean("mqProducerConfig")
    @ConfigurationProperties(prefix = "rocketmq.producer")
    public MQProducerProperties mqProducerConfig(){
        return new MQProducerProperties();
    }

    @Bean("defaultMQProducer")
    public DefaultMQProducer defaultMQProducer(MQProducerProperties mqProducerConfig) throws MQClientException {
        DefaultMQProducer mqProducer  = null;
        if("Y".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff()) || "true".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff()) || "on".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff())){
            //是否开启acl
            String aclSecretKey = SM4Util.sm4dec("B001",mqProducerConfig.getAclSecretKey());
            AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqProducerConfig.getAclAccessKey(), aclSecretKey ));
            //AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqProducerConfig.getAclAccessKey(), mqProducerConfig.getAclSecretKey()));
            mqProducer  = new DefaultMQProducer(mqProducerConfig.getGroupName(),aclClientRPCHook);
        }else {
            mqProducer  = new DefaultMQProducer(mqProducerConfig.getGroupName());
        }
        mqProducer.setNamesrvAddr(mqProducerConfig.getNameSrvAddr());
        mqProducer.setVipChannelEnabled(false);
        if(mqProducerConfig.getMaxMessageSize()>0){
            mqProducer.setMaxMessageSize(mqProducerConfig.getMaxMessageSize());
        }
        mqProducer.setSendMsgTimeout(mqProducerConfig.getSendMsgTimeOut());
        mqProducer.setRetryTimesWhenSendAsyncFailed(mqProducerConfig.getRetryTimesWhenSendFailed());

        mqProducer.start();
        return mqProducer;
    }

    @Bean("transactionMQProducer ")
    public TransactionMQProducer transactionMQProducer(MQProducerProperties mqProducerConfig) throws MQClientException {
        TransactionMQProducer mqProducer  = null;
        //实例化事务监听器
        TransactionListener transactionListener = new TransactionListenerImpl();
        if("Y".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff()) || "true".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff()) || "on".equalsIgnoreCase(mqProducerConfig.getAclIsOnOff())){
            //是否开启acl
            String aclSecretKey = SM4Util.sm4dec("B001",mqProducerConfig.getAclSecretKey());
            AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqProducerConfig.getAclAccessKey(), aclSecretKey ));
            //AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqProducerConfig.getAclAccessKey(), mqProducerConfig.getAclSecretKey()));
            mqProducer  = new TransactionMQProducer(mqProducerConfig.getGroupName(),aclClientRPCHook);
        }else {
            mqProducer  = new TransactionMQProducer(mqProducerConfig.getGroupName());
        }
        mqProducer.setNamesrvAddr(mqProducerConfig.getNameSrvAddr());

        //创建线程池
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("client-transaction-msg-check-thread");
                        return thread;
                    }
                });
        //producer设置线程池 和 事务监听器
        mqProducer.setExecutorService(executorService);
        mqProducer.setTransactionListener(transactionListener);
        mqProducer.setVipChannelEnabled(false);
        if(mqProducerConfig.getMaxMessageSize()>0){
            mqProducer.setMaxMessageSize(mqProducerConfig.getMaxMessageSize());
        }
        mqProducer.setSendMsgTimeout(mqProducerConfig.getSendMsgTimeOut());
        mqProducer.setRetryTimesWhenSendAsyncFailed(mqProducerConfig.getRetryTimesWhenSendFailed());

        mqProducer.start();
        return mqProducer;
    }

}
